package commands;

import java.io.IOException;
import java.util.Arrays;

import mapReduce.ListCommentsMR;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;

import vo.Comment;

public class ListCommentsCommand implements Comando {

	private static byte[] ALL_ZEROS = new byte[Bytes.SIZEOF_BYTE];
	private static byte[] ALL_ONES = new byte[Bytes.SIZEOF_BYTE];

	static {
		Arrays.fill(ALL_ZEROS, (byte) 0x00);
		Arrays.fill(ALL_ONES, (byte) 0xFF);
	}

	public String execute(String[] args) throws IOException {

		if (args.length != 4) {
			return getSintaxe();
		}

		Configuration conf = HBaseConfiguration.create();

		try {
			createJob(conf, args).waitForCompletion(true);
		} catch (InterruptedException e) {
			e.printStackTrace();

		} catch (ClassNotFoundException e) {
			e.printStackTrace();

		}

		return ListCommentsMR.Reducer.printListComments();
	}

	public Job createJob(Configuration conf, String[] args) throws IOException {
		Job job = new Job(conf, "teste");
		job.setJarByClass(this.getClass());
		String table = Comment.TABLENAME;

		byte[] photoIdInBytes = Bytes.toBytes(args[1]);
		Integer pageSize = Integer.valueOf(args[2]);
		Integer pageNumber = Integer.valueOf(args[3]);

		Scan scan = new Scan();
		scan.setFilter(new PageFilter(pageSize));

		scan.setStartRow(Bytes.add(photoIdInBytes, ALL_ZEROS));
		scan.setStopRow(Bytes.add(photoIdInBytes, ALL_ONES));
		scan.addColumn(Bytes.toBytes(Comment.CF_ATTRIBUTES), Bytes
				.toBytes(Comment.COMMENT_TEXT));

		job.setNumReduceTasks(1);

		TableMapReduceUtil.initTableMapperJob(table, scan,
				ListCommentsMR.Mapper.class, ImmutableBytesWritable.class,
				ImmutableBytesWritable.class, job);

		TableMapReduceUtil.initTableReducerJob(table,
				ListCommentsMR.Reducer.class, job);

		return job;
	}

	@Override
	public String getSintaxe() {
		return "-listComments photoId pageNumber pageSize";
	}

	@Override
	public String getKey() {
		return "-listComments";
	}

}